package dk.almbrand.microservice.kafka.producer.secure;

import dk.almbrand.microservice.circuitbreaker.CircuitBreaker;

import dk.almbrand.microservice.kafka.producer.Producer;

import java.util.concurrent.Callable;

public class ProtectedProducerImpl implements Callable, ProtectedProducer {

    private String topic; 
    private String key;
    private String message;
                                         
    private Producer producer;
    private CircuitBreaker circuitBreaker;
    
    public ProtectedProducerImpl(Producer producer, CircuitBreaker circuitBreaker) {
        this.producer = producer;
        this.circuitBreaker = circuitBreaker;
    }
    
    public void publish(String topic, String key, String message) {
        if(circuitBreaker != null) {
            // Temporary storing of values for when call is executed from the circuit breaker.
            this.topic = topic;
            this.key = key;
            this.message = message;
            
            circuitBreaker.doAction(this, Object.class);
        } 
    }

    @Override
    public Object call() throws Exception {
        System.out.println("i call data: " + message + " topic: " + topic + " key: " + key);
        producer.publish(topic, key, message);
        return null;
    }
}
